package com.aispeech.asr.apis.detection.controller;

import java.io.IOException;
import java.io.InputStream;
import java.util.UUID;
import java.util.function.BiConsumer;
import java.util.function.Supplier;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.core.io.buffer.DataBuffer;
import org.springframework.core.io.buffer.NettyDataBuffer;
import org.springframework.core.io.buffer.NettyDataBufferFactory;
import org.springframework.http.MediaType;
import org.springframework.http.codec.multipart.FilePart;
import org.springframework.http.codec.multipart.Part;
import org.springframework.web.bind.annotation.CrossOrigin;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestParam;
import org.springframework.web.bind.annotation.RequestPart;
import org.springframework.web.bind.annotation.RestController;

import com.aispeech.asr.apis.detection.domain.Audio;
import com.aispeech.asr.apis.detection.exceptions.BusinessException;
import com.aispeech.asr.apis.detection.form.DetectionForm;
import com.aispeech.asr.apis.detection.service.AsrService;
import com.aispeech.asr.comm.data.vo.Result;
import com.aispeech.asr.comm.enums.ErrorCode;
import com.aispeech.asr.webase.controller.Base;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.fasterxml.jackson.databind.exc.InvalidFormatException;
import com.fasterxml.jackson.databind.exc.UnrecognizedPropertyException;

import lombok.extern.slf4j.Slf4j;
import reactor.core.publisher.Mono;
import reactor.util.function.Tuple2;

@RestController
@CrossOrigin
@Slf4j
public class DetectionController extends Base {
    @Autowired
    ObjectMapper objectMapper;

    @Autowired
    AsrService service;

    @Autowired
    NettyDataBufferFactory nettyDataBufferFactory;

    @PostMapping(value = "/sed-api/v1/detection", consumes = MediaType.MULTIPART_FORM_DATA_VALUE)
    public Mono<Result> post(@RequestPart("params") Part paramsPart,
                             @RequestPart("file") FilePart filePart,
                             @RequestParam("productId") String productId) {

        return Mono
                .zip(
                        paramsPart.content().collect(new DataBufferJoiner(), new DataBufferMerger()),
                        filePart.content().collect(new DataBufferJoiner(), new DataBufferMerger())
                ).flatMap(tuple -> onReceived(productId, tuple));
    }

    private Mono<Result> onReceived(String productId, Tuple2<NettyDataBuffer, NettyDataBuffer> tuple) {
        NettyDataBuffer formBuffer = tuple.getT1();
        NettyDataBuffer audioBuffer = tuple.getT2();

        try {
        	DetectionForm form = objectMapper.readValue(formBuffer.asInputStream(), DetectionForm.class);
        	Audio audio = form.getAudio();
            if (audio == null || audio.getAudioType() == null || audio.getSampleRate() == 0) {
                log.warn("param 'audio' error. form: {}", form);
                return Mono.just(Result.fail(ErrorCode.ERR_PARAMS_MISSED.errid, "missing param 'audio' or parsing failed"));
            }
            
            if (!"wav".equalsIgnoreCase(audio.getAudioType()) || audio.getSampleRate() != 16000 || audio.getChannel() != 1 || audio.getSampleBytes() != 2) {
                return Mono.just(Result.fail(ErrorCode.ERR_PARAMS_MISSED.errid, "sorry, current version only supports 16000 SampleRate, 2 SampleBytes, 1 Channel WAV audio"));
            }
            
            int audioLength = audioBuffer.readableByteCount();
            form.setProductId(productId);
            form.setDuration(pcmBytesToMsecs(audioLength, audio.getSampleRate(), audio.getChannel(), audio.getSampleBytes()));
            form.setLength(audioLength);
        	InputStream inputStream = audioBuffer.asInputStream(true);
            String sessionId = UUID.randomUUID().toString().replaceAll("-", "");
            log.info("[ onReceived ]: sessionId: {}, form: {}", sessionId, form);

//            File fileDest = new File(String.format("/audios/%s/%s.%s",
//                    LocalDateTime.now().format(DateTimeFormatter.ofPattern("yyyyMMdd")),
//                    sessionId,
//                    form.getAudio().getAudioType()));
//            if (!fileDest.getParentFile().exists()) fileDest.getParentFile().mkdir();
//            Files.copy(inputStream, Paths.get(fileDest.toURI()));
            
            
            return service
                    .process(sessionId, form, inputStream)
                    .doOnTerminate(() -> {
//                    	audioBuffer.release();
//                    	try {
//							Files.delete(Paths.get(fileDest.toURI()));
//						} catch (IOException e) {
//							// TODO Auto-generated catch block
//							log.error("[ Files.delete ]: sessionId: {}, url: {}", sessionId, fileDest.toURI().toString());
//						}
                    }
                    );
        } catch (IOException ex) {
            audioBuffer.release();
            String str = "";
            //log.error("[onReceived] parm parsing failed",ex);

            if (ex instanceof UnrecognizedPropertyException) {
                UnrecognizedPropertyException unrecognizedPropertyException = (UnrecognizedPropertyException) ex;
                str = unrecognizedPropertyException.getMessage();
            } else if(ex instanceof InvalidFormatException){
            	InvalidFormatException invalidFormatException = (InvalidFormatException) ex;
                 str = invalidFormatException.getMessage();
            }
            else {
                str = "parsing failed";
            }

            return Mono.error(new BusinessException(ErrorCode.ERR_PARAMS_MISSED.errid, str.split("\n")[0]));
        } finally {
            formBuffer.release();
        }
    }

    class DataBufferJoiner implements Supplier<NettyDataBuffer> {
        @Override
        public NettyDataBuffer get() {
            return nettyDataBufferFactory.allocateBuffer();
        }
    }

    class DataBufferMerger implements BiConsumer<NettyDataBuffer, DataBuffer> {

        @Override
        public void accept(NettyDataBuffer nettyDataBuffer, DataBuffer buffer) {
            nettyDataBuffer.write(buffer);
        }
    }
    
    private long pcmBytesToMsecs(long offset, int sampleRate, int channel, int sampleBytes) {
        return offset * 1000 / (sampleRate * channel * sampleBytes);
    }
}
